\section{Implementation}\label{sec:arch}



\paragraph{Architecture.}
 
\name\ uses the heuristics defined previously to form optimal hierarchies for compute-aggregate problems.  
 Figure~\ref{fig:Arch} overviews its architecture. Computation nodes
 are managed by a big data analysis system (e.g., Apache
 Crunch~\cite{ApacheCrunch} or Spark~\cite{rdds}) which uses \name.
 When a compute-aggregate problem is initialized, computation nodes start the first phase. A list of
 the computation nodes and the traits of the aggregation function are sent to the
 Controller within \name. After this \name\ determines the
 optimal fanout and determines the appropriate overlay on an ordered set of
 vertices. The controller tells each vertex its children and parent. All nodes
 create the relevant TCP connections while waiting for the computation phase to
 complete.
 
 When a computation node finishes, it sends its results to the appropriate
 vertex.
 Each aggregator waits until it has results from all its children, then
 aggregates and pushes the results to its parent. When the root finishes it can
 hand the result over to %back to a specified node in
 the big data analysis system or write it out to a distributed file system. If
 the overlay is created for ongoing computation then the vertices maintain their
 connections and do work when new input is pushed in. Otherwise a node drops
 connections after communicating its portion to its parent.
 
 \begin{figure}[!ht]
 \centering
 \includegraphics[width=0.9\columnwidth, keepaspectratio=true,trim=0 0 0 0,
 clip]{Architecture}
 \caption{Architecture}
 \label{fig:Arch}
 \end{figure}


\paragraph{Programming abstractions.}
  
Next we outline a minor API extension to FlumeJava~\cite{FlumeJava} implemented in Apache Crunch~\cite{ApacheCrunch} for users to 
directly, more easily, define compute-aggregate operations. FlumeJava is a Java library for dealing with collections that represent possibly large amounts of data, and expressing operations that should be performed on these collections. The two main data structures of FlumeJava are \lstinline{PCollection} and \lstinline{PTable}, representing a collection of elements and a collection of \emph{key-value} pairs %of the same type(s) 
respectively. FlumeJava proposes four main types of operations that can be performed on its data structures: (1) \lstinline{parallelDo()} performs a general operation defined by a function object (\lstinline{DoFn}) on each element; (2) \lstinline{groupByKey()} groups elements of a \lstinline{PTable} based on keys; (3) \lstinline{combineVales()} combines elements of a grouped \lstinline{PTable} that has the same key based on a function object of type \lstinline{CombineFn}; (4) \lstinline{flatten()} flattens elements of two or more \lstinline{PCollection}s resulting in a single \lstinline{PCollection}.  

We extend FlumeJava with a fifth main operation \lstinline{parallelAggregate()}. The syntax of the operation performed on a \lstinline{PCollection<A>} is given below.
      
\begin{lstlisting}
PCollection<B> parallelAggregate(DoFn<A,B>, 
	AggregateFn<B>) 
\end{lstlisting}

\lstinline{parallelAggregate()} takes two parameters: (1) a compute function of type \lstinline{DoFn<A,B>} and (2) an aggregation function of type \lstinline{AggregateFn<B>}. The result of the operation is a \lstinline{PCollection<B>} that represents the aggregated data. The following example illustrates how  \lstinline{parallelAggregate()} can be used to concisely determine the top-$k$ elements of a dataset.

\begin{lstlisting} 
PCollection<String> originalElements = ...;	
PCollection<Integer> topK = 
  originalElements.parallelAggregate(
    computeFn, aggregateFn);
\end{lstlisting}

For brevity we elide the computation function \lstinline{computeFn} used to generate \lstinline{Integer} elements from a  \lstinline{String} dataset and the aggregation function \lstinline{aggregateFn} used to determine the top-$k$ elements of the set of \lstinline{Integer}s. The operation returns a \lstinline{PCollection} that contains the top-$k$ elements.  

